package cn.lzd.mr.case7;

import cn.lzd.mr.PathUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.Iterator;

/**
 * Created by lzd on 2018/6/5.
 * join算法 ，将两张表的字段合并，关联字段userid
 * 效率底下
 */
public class Case7MR2 {

    static class Mp extends Mapper<LongWritable, Text, Bean2, NullWritable> {

        String filename;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            FileSplit fileSplit = (FileSplit) context.getInputSplit();
            filename = fileSplit.getPath().getName();
            System.out.println("====================hello=======================" + filename);
        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            System.out.println("====================================map 开始====================================");
            String[] strs = value.toString().split(",");

            if (filename.indexOf("order") != -1) {
                Bean2 bean = new Bean2(strs[0], strs[1], "null", -1, "null", filename);
                context.write(bean, NullWritable.get());
            } else {
                Bean2 bean2 = new Bean2("null", strs[0], strs[1], Integer.parseInt(strs[2]), strs[3], filename);
                context.write(bean2, NullWritable.get());
            }
            System.out.println("====================================map 结束====================================");
        }
    }

    static class Rd extends Reducer<Bean2, NullWritable, Bean2, NullWritable> {
        Bean2 orderBean = new Bean2();
        Bean2 userBean = new Bean2();

        @Override
        protected void reduce(Bean2 bean, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {

            System.out.println("====================================reduce 开始====================================");
            try {
//                for (NullWritable v : values) {
//                    if("user".equals(bean.getFilename())) {
//
//                        BeanUtils.copyProperties(userBean, bean);
//
//                    }else{
//                        BeanUtils.copyProperties(orderBean, bean);
//                        orderBean.setName(userBean.getName());
//                        orderBean.setAge(userBean.getAge());
//                        orderBean.setStageName(userBean.getStageName());
//                        context.write(orderBean,NullWritable.get());
//                    }
//                }
                Iterator<NullWritable> iterator = values.iterator();
                while (iterator.hasNext()) {
                    iterator.next();
                }
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }


//            int i = 0;
//            for (NullWritable v : values) {
//                System.out.println(bean + "============" + (++i));
//                context.write(bean, NullWritable.get());
//            }
            System.out.println("====================================reduce 结束====================================");
        }
    }

    public static void main(String[] args) throws Exception {

        String inpath = PathUtil.getpath("case7") + "/*";
        String outpath = PathUtil.getpath("case7") + "out2/";

        Configuration conf = new Configuration();

        PathUtil.handleOutPath(outpath, conf);

        Job job = Job.getInstance(conf);

        job.setPartitionerClass(CustomParition.class);
        job.setNumReduceTasks(2);
        job.setGroupingComparatorClass(GroupComparator.class);

        job.setJarByClass(Case7MR2.class);

        job.setMapperClass(Case7MR2.Mp.class);
        job.setReducerClass(Case7MR2.Rd.class);

        job.setMapOutputKeyClass(Bean2.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(Bean2.class);
        job.setOutputValueClass(NullWritable.class);


        FileInputFormat.addInputPath(job, new Path(inpath));
        FileOutputFormat.setOutputPath(job, new Path(outpath));


        job.waitForCompletion(true);

    }
}
